package cn.spark.study.core;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.sources.In;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

/**
 * transformation案例实战
 */
public class TransformationOperation {
    public static void main(String[] args) {
        // map();
        // filter();
        // flatMap();
        // groupByKey();
        // reduceByKey();
        // sortByKey();
        // join();
        cogroup();
    }

    private static void map() {
        SparkConf conf = new SparkConf().setAppName("map").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);

        JavaRDD<Integer> numbersRDD = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5));

        JavaRDD<Integer> resultRDD = numbersRDD.map(new Function<Integer, Integer>() {
            @Override
            public Integer call(Integer v1) throws Exception {
                System.out.println("v1 = " + v1);
                return v1 * 2;
            }
        });

        resultRDD.foreach(new VoidFunction<Integer>() {
            @Override
            public void call(Integer integer) throws Exception {
                System.out.println(integer);
            }
        });

        sc.close();
    }

    private static void filter() {
        SparkConf conf = new SparkConf().setAppName("filter").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);

        List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        JavaRDD<Integer> numbersRDD = sc.parallelize(numbers);
        JavaRDD evenNumberRDD = numbersRDD.filter(new Function<Integer, Boolean>() {
            @Override
            public Boolean call(Integer integer) throws Exception {
                return integer % 2 == 0;
            }
        });

        evenNumberRDD.foreach(new VoidFunction<Integer>() {
            @Override
            public void call(Integer v) throws Exception {
                System.out.println("v = " + v);
            }
        });

        sc.close();
    }

    /**
     * flatMap案例：将文本行拆分成多个单词
     */
    private static void flatMap() {
        // 创建SparkConf
        SparkConf conf = new SparkConf().setAppName("flatMap").setMaster("local");
        // 创建JavaSparkContext
        JavaSparkContext sc = new JavaSparkContext(conf);
        // 并行化集合，创建RDD
        List<String> lineList = Arrays.asList("hello you", "hello me", "hello word");
        JavaRDD<String> lines = sc.parallelize(lineList);
        // 使用flatMap算子，将每行文本拆分为每个单词
        JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterable<String> call(String line) throws Exception {
                return Arrays.asList(line.split(" "));
            }
        });

        // 打印新的RDD
        words.foreach(new VoidFunction<String>() {
            @Override
            public void call(String word) throws Exception {
                System.out.println(word);
            }
        });

        // 关闭Java Spark Context
        sc.close();
    }


    /**
     * groupByKey案例：按照班级对成绩分组
     */
    private static void groupByKey() {
        // 创建SparkConf
        SparkConf conf = new SparkConf().setAppName("groupByKey").setMaster("local");
        // 创建JavaSparkContext
        JavaSparkContext sc = new JavaSparkContext(conf);

        // 模拟集合
        List<Tuple2<String, Integer>> scoreList = Arrays.asList(
                new Tuple2<String, Integer>("class1", 80),
                new Tuple2<String, Integer>("class2", 75),
                new Tuple2<String, Integer>("class1", 90),
                new Tuple2<String, Integer>("class2", 65)
        );

        // 并行化集合 创建JavaPairRDD
        JavaPairRDD<String, Integer> scores = sc.parallelizePairs(scoreList);

        // 对scores RDD 执行 groupByKey算子，对每个班级的成绩进行分组
        JavaPairRDD<String, Iterable<Integer>> groupedScores = scores.groupByKey();

        groupedScores.foreach(new VoidFunction<Tuple2<String, Iterable<Integer>>>() {
            @Override
            public void call(Tuple2<String, Iterable<Integer>> tuple2) throws Exception {
                System.out.println("班级：" + tuple2._1);
                Iterator<Integer> iterator = tuple2._2.iterator();
                while(iterator.hasNext()) {
                    System.out.println(iterator.next());
                }
                System.out.println("======================================");
            }
        });

        // 关闭Java Spark Context
        sc.close();
    }

    /**
     * reduceByKey案例：统计每个班级的总分
     */
    private static void reduceByKey() {
        // 创建SparkConf
        SparkConf conf = new SparkConf().setAppName("reduceByKey").setMaster("local");
        // 创建JavaSparkContext
        JavaSparkContext sc = new JavaSparkContext(conf);

        // 模拟集合
        List<Tuple2<String, Integer>> scoreList = Arrays.asList(
                new Tuple2<String, Integer>("class1", 80),
                new Tuple2<String, Integer>("class2", 75),
                new Tuple2<String, Integer>("class1", 90),
                new Tuple2<String, Integer>("class2", 65)
        );

        // 并行化集合 创建JavaPairRDD
        JavaPairRDD<String, Integer> scores = sc.parallelizePairs(scoreList);

        // 对scores RDD 执行 reduceByKey算子，对每个班级的成绩进行分组
        JavaPairRDD<String, Integer> totalScores = scores.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });

        //打印 totalScores RDD
        totalScores.foreach(new VoidFunction<Tuple2<String, Integer>>() {
            @Override
            public void call(Tuple2<String, Integer> t) throws Exception {
                System.out.println(t._1 + " : " + t._2);
            }
        });

        // 关闭Java Spark Context
        sc.close();
    }

    /**
     * sortByKey案例：按照学生分数排序
     */
    private static void sortByKey() {
        // 创建SparkConf
        SparkConf conf = new SparkConf().setAppName("sortByKey").setMaster("local");
        // 创建JavaSparkContext
        JavaSparkContext sc = new JavaSparkContext(conf);

        // 模拟集合
        List<Tuple2<Integer, String>> scoreList = Arrays.asList(
                new Tuple2<Integer, String>(65, "leo"),
                new Tuple2<Integer, String>(50, "tom"),
                new Tuple2<Integer, String>(100, "marry"),
                new Tuple2<Integer, String>(80, "jack")
        );

        // 并行化集合 创建JavaPairRDD
        JavaPairRDD<Integer, String> scores = sc.parallelizePairs(scoreList);

        // 对scores RDD 执行 sortedByKey算子
        JavaPairRDD<Integer, String> sortedScores = scores.sortByKey(false);

        //打印 sortedScores RDD
        sortedScores.foreach(new VoidFunction<Tuple2<Integer, String>>() {
            @Override
            public void call(Tuple2<Integer, String> t) throws Exception {
                System.out.println(t._2 + " : " + t._1);
            }
        });

        // 关闭Java Spark Context
        sc.close();
    }

    /**
     * join 案例：打印学生成绩
     */
    private static void join() {
        // 创建SparkConf
        SparkConf conf = new SparkConf().setAppName("join").setMaster("local");
        // 创建JavaSparkContext
        JavaSparkContext sc = new JavaSparkContext(conf);

        // 模拟集合
        List<Tuple2<Integer, String>> studentList = Arrays.asList(
                new Tuple2<Integer, String>(1, "leo"),
                new Tuple2<Integer, String>(2, "marry"),
                new Tuple2<Integer, String>(3, "jack")
        );
        List<Tuple2<Integer, Integer>> scoreList = Arrays.asList(
                new Tuple2<Integer, Integer>(1, 85),
                new Tuple2<Integer, Integer>(2, 100),
                new Tuple2<Integer, Integer>(3, 60)
        );

        // 并行化两个集合
        JavaPairRDD<Integer, String> students = sc.parallelizePairs(studentList);
        JavaPairRDD<Integer, Integer> scores = sc.parallelizePairs(scoreList);

        // 使用join算子关联两个RDD
        JavaPairRDD<Integer, Tuple2<String, Integer>> studentScores = students.join(scores);

        // 打印studentScores RDD
        studentScores.foreach(new VoidFunction<Tuple2<Integer, Tuple2<String, Integer>>>() {
            @Override
            public void call(Tuple2<Integer, Tuple2<String, Integer>> t) throws Exception {
                System.out.println("编号：" + t._1 );
                System.out.println("姓名：" + t._2._1 );
                System.out.println("成绩：" + t._2._2 );
                System.out.println("=============================");
            }
        });



        // 关闭Java Spark Context
        sc.close();
    }


    /**
     * cogroup 案例：打印学生成绩
     */
    private static void cogroup() {
        // 创建SparkConf
        SparkConf conf = new SparkConf().setAppName("cogroup").setMaster("local");
        // 创建JavaSparkContext
        JavaSparkContext sc = new JavaSparkContext(conf);

        // 模拟集合
        List<Tuple2<Integer, String>> studentList = Arrays.asList(
                new Tuple2<Integer, String>(1, "leo"),
                new Tuple2<Integer, String>(2, "marry"),
                new Tuple2<Integer, String>(3, "jack")
        );
        List<Tuple2<Integer, Integer>> scoreList = Arrays.asList(
                new Tuple2<Integer, Integer>(1, 85),
                new Tuple2<Integer, Integer>(2, 100),
                new Tuple2<Integer, Integer>(3, 60),
                new Tuple2<Integer, Integer>(1, 75),
                new Tuple2<Integer, Integer>(2, 90),
                new Tuple2<Integer, Integer>(3, 50)
        );

        // 并行化两个集合
        JavaPairRDD<Integer, String> students = sc.parallelizePairs(studentList);
        JavaPairRDD<Integer, Integer> scores = sc.parallelizePairs(scoreList);

        // 使用cogroup算子关联两个RDD
        JavaPairRDD<Integer, Tuple2<Iterable<String>, Iterable<Integer>>> studentScores = students.cogroup(scores);

        // 打印studentScores RDD
        studentScores.foreach(new VoidFunction<Tuple2<Integer, Tuple2<Iterable<String>, Iterable<Integer>>>>() {
            @Override
            public void call(Tuple2<Integer, Tuple2<Iterable<String>, Iterable<Integer>>> t) throws Exception {
                System.out.println("编号：" + t._1 );
                System.out.println("姓名：" + t._2._1 );
                System.out.println("成绩：" + t._2._2 );
                System.out.println("=============================");
            }
        });

        // 关闭Java Spark Context
        sc.close();
    }

}
